/* http.cc
   Mathieu Stefani, 13 August 2015

   Http layer implementation
*/

#include <pistache/config.h>
#include <pistache/http.h>
#include <pistache/net.h>
#include <pistache/peer.h>
#include <pistache/transport.h>

#include <cstring>
#include <ctime>
#include <iomanip>
#include <iostream>
#include <stdexcept>
#include <unordered_map>

#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

namespace Pistache {
namespace Http {

template <typename H, typename Stream, typename... Args>
typename std::enable_if<Header::IsHeader<H>::value, Stream &>::type
writeHeader(Stream &stream, Args &&... args) {
  H header(std::forward<Args>(args)...);

  stream << H::Name << ": ";
  header.write(stream);

  stream << crlf;

  return stream;
}

namespace {
bool writeStatusLine(Version version, Code code, DynamicStreamBuf &buf) {
#define OUT(...)                                                               \
  do {                                                                         \
    __VA_ARGS__;                                                               \
    if (!os)                                                                   \
      return false;                                                            \
  } while (0)

  std::ostream os(&buf);

  OUT(os << version << " ");
  OUT(os << static_cast<int>(code));
  OUT(os << ' ');
  OUT(os << code);
  OUT(os << crlf);

  return true;

#undef OUT
}

bool writeHeaders(const Header::Collection &headers, DynamicStreamBuf &buf) {
#define OUT(...)                                                               \
  do {                                                                         \
    __VA_ARGS__;                                                               \
    if (!os)                                                                   \
      return false;                                                            \
  } while (0)

  std::ostream os(&buf);

  for (const auto &header : headers.list()) {
    OUT(os << header->name() << ": ");
    OUT(header->write(os));
    OUT(os << crlf);
  }

  return true;

#undef OUT
}

bool writeCookies(const CookieJar &cookies, DynamicStreamBuf &buf) {
#define OUT(...)                                                               \
  do {                                                                         \
    __VA_ARGS__;                                                               \
    if (!os)                                                                   \
      return false;                                                            \
  } while (0)

  std::ostream os(&buf);
  for (const auto &cookie : cookies) {
    OUT(os << "Set-Cookie: ");
    OUT(os << cookie);
    OUT(os << crlf);
  }

  return true;

#undef OUT
}

using HttpMethods = std::unordered_map<std::string, Method>;

const HttpMethods httpMethods = {
#define METHOD(repr, str) {str, Method::repr},
    HTTP_METHODS
#undef METHOD
};

} // namespace

static constexpr const char *ParserData = "__Parser";

namespace Private {

Step::Step(Message *request) : message(request) {}

void Step::raise(const char *msg, Code code /* = Code::Bad_Request */) {
  throw HttpError(code, msg);
}

State RequestLineStep::apply(StreamCursor &cursor) {
  StreamCursor::Revert revert(cursor);

  auto request = static_cast<Request *>(message);

  StreamCursor::Token methodToken(cursor);
  if (!match_until(' ', cursor))
    return State::Again;

  auto it = httpMethods.find(methodToken.text());
  if (it != httpMethods.end()) {
    request->method_ = it->second;
  } else {
    raise("Unknown HTTP request method");
  }

  int n;

  if (cursor.eof())
    return State::Again;
  else if ((n = cursor.current()) != ' ')
    raise("Malformed HTTP request after Method, expected SP");

  if (!cursor.advance(1))
    return State::Again;

  StreamCursor::Token resToken(cursor);
  while ((n = cursor.current()) != '?' && n != ' ')
    if (!cursor.advance(1))
      return State::Again;

  request->resource_ = resToken.text();

  // Query parameters of the Uri
  if (n == '?') {
    if (!cursor.advance(1))
      return State::Again;

    while ((n = cursor.current()) != ' ') {
      StreamCursor::Token keyToken(cursor);
      if (!match_until({'=', ' ', '&'}, cursor))
        return State::Again;

      std::string key = keyToken.text();

      auto c = cursor.current();
      if (c == ' ') {
        request->query_.add(std::move(key), "");
      } else if (c == '&') {
        request->query_.add(std::move(key), "");
        if (!cursor.advance(1))
          return State::Again;
      } else if (c == '=') {
        if (!cursor.advance(1))
          return State::Again;

        StreamCursor::Token valueToken(cursor);
        if (!match_until({' ', '&'}, cursor))
          return State::Again;

        std::string value = valueToken.text();
        request->query_.add(std::move(key), std::move(value));
        if (cursor.current() == '&') {
          if (!cursor.advance(1))
            return State::Again;
        }
      }
    }
  }

  // @Todo: Fragment

  // SP
  if (!cursor.advance(1))
    return State::Again;

  // HTTP-Version
  StreamCursor::Token versionToken(cursor);

  while (!cursor.eol())
    if (!cursor.advance(1))
      return State::Again;

  const char *ver = versionToken.rawText();
  const size_t size = versionToken.size();
  if (strncmp(ver, "HTTP/1.0", size) == 0) {
    request->version_ = Version::Http10;
  } else if (strncmp(ver, "HTTP/1.1", size) == 0) {
    request->version_ = Version::Http11;
  } else {
    raise("Encountered invalid HTTP version");
  }

  if (!cursor.advance(2))
    return State::Again;

  revert.ignore();
  return State::Next;
}

State ResponseLineStep::apply(StreamCursor &cursor) {
  StreamCursor::Revert revert(cursor);

  auto *response = static_cast<Response *>(message);

  if (match_raw("HTTP/1.1", strlen("HTTP/1.1"), cursor)) {
    // response->version = Version::Http11;
  } else if (match_raw("HTTP/1.0", strlen("HTTP/1.0"), cursor)) {
  } else {
    raise("Encountered invalid HTTP version");
  }

  int n;
  // SP
  if ((n = cursor.current()) != StreamCursor::Eof && n != ' ')
    raise("Expected SPACE after http version");
  if (!cursor.advance(1))
    return State::Again;

  StreamCursor::Token codeToken(cursor);
  if (!match_until(' ', cursor))
    return State::Again;

  char *end;
  auto code = strtol(codeToken.rawText(), &end, 10);
  if (*end != ' ')
    raise("Failed to parsed return code");
  response->code_ = static_cast<Http::Code>(code);

  if (!cursor.advance(1))
    return State::Again;

  while (!cursor.eol() && !cursor.eof()) {
    cursor.advance(1);
  }

  if (!cursor.advance(2))
    return State::Again;

  revert.ignore();
  return State::Next;
}

State HeadersStep::apply(StreamCursor &cursor) {
  StreamCursor::Revert revert(cursor);

  while (!cursor.eol()) {
    StreamCursor::Revert headerRevert(cursor);

    // Read the header name
    size_t start = cursor;

    while (cursor.current() != ':')
      if (!cursor.advance(1))
        return State::Again;

    // Skip the ':'
    if (!cursor.advance(1))
      return State::Again;

    std::string name =
        std::string(cursor.offset(start), cursor.diff(start) - 1);

    // Ignore spaces
    while (cursor.current() == ' ')
      if (!cursor.advance(1))
        return State::Again;

    // Read the header value
    start = cursor;
    while (!cursor.eol()) {
      if (!cursor.advance(1))
        return State::Again;
    }

    if (Header::LowercaseEqualStatic(name, "cookie")) {
      message->cookies_.removeAllCookies(); // removing existing cookies before
                                            // re-adding them.
      message->cookies_.addFromRaw(cursor.offset(start), cursor.diff(start));
    } else if (Header::LowercaseEqualStatic(name, "set-cookie")) {
      message->cookies_.add(
          Cookie::fromRaw(cursor.offset(start), cursor.diff(start)));
    }

    // If the header is registered with the Registry, add its strongly
    //  typed form to the headers list...
    else if (Header::Registry::instance().isRegistered(name)) {
      std::shared_ptr<Header::Header> header =
          Header::Registry::instance().makeHeader(name);
      header->parseRaw(cursor.offset(start), cursor.diff(start));
      message->headers_.add(header);
    }

    // But also preserve a raw header version too, regardless of whether
    //  its type was known to the Registry...
    std::string value(cursor.offset(start), cursor.diff(start));
    message->headers_.addRaw(Header::Raw(std::move(name), std::move(value)));

    // CRLF
    if (!cursor.advance(2))
      return State::Again;

    headerRevert.ignore();
  }

  if (!cursor.advance(2))
    return State::Again;

  revert.ignore();
  return State::Next;
}

State BodyStep::apply(StreamCursor &cursor) {
  auto cl = message->headers_.tryGet<Header::ContentLength>();
  auto te = message->headers_.tryGet<Header::TransferEncoding>();

  if (cl && te)
    raise("Got mutually exclusive ContentLength and TransferEncoding header");

  if (cl)
    return parseContentLength(cursor, cl);

  if (te)
    return parseTransferEncoding(cursor, te);

  return State::Done;
}

State BodyStep::parseContentLength(
    StreamCursor &cursor, const std::shared_ptr<Header::ContentLength> &cl) {
  auto contentLength = cl->value();

  auto readBody = [&](size_t size) {
    StreamCursor::Token token(cursor);
    const size_t available = cursor.remaining();

    // We have an incomplete body, read what we can
    if (available < size) {
      cursor.advance(available);
      message->body_.append(token.rawText(), token.size());

      bytesRead += available;

      return false;
    }

    cursor.advance(size);
    message->body_.append(token.rawText(), token.size());
    return true;
  };

  // We already started to read some bytes but we got an incomplete payload
  if (bytesRead > 0) {
    // How many bytes do we still need to read ?
    const size_t remaining = contentLength - bytesRead;
    if (!readBody(remaining))
      return State::Again;
  }
  // This is the first time we are reading the payload
  else {
    message->body_.reserve(contentLength);
    if (!readBody(contentLength))
      return State::Again;
  }

  bytesRead = 0;
  return State::Done;
}

BodyStep::Chunk::Result BodyStep::Chunk::parse(StreamCursor &cursor) {
  if (size == -1) {
    StreamCursor::Revert revert(cursor);
    StreamCursor::Token chunkSize(cursor);

    while (!cursor.eol())
      if (!cursor.advance(1))
        return Incomplete;

    char *end;
    const char *raw = chunkSize.rawText();
    auto sz = std::strtol(raw, &end, 16);
    if (*end != '\r')
      throw std::runtime_error("Invalid chunk size");

    // CRLF
    if (!cursor.advance(2))
      return Incomplete;

    revert.ignore();

    size = sz;
  }

  if (size == 0)
    return Final;

  message->body_.reserve(size);
  StreamCursor::Token chunkData(cursor);
  const ssize_t available = cursor.remaining();

  if (static_cast<ssize_t>(available + message->body_.size()) < size) {
    cursor.advance(available);
    message->body_.append(chunkData.rawText(), available);
    return Incomplete;
  }
  cursor.advance(size - message->body_.size());

  if (!cursor.advance(2))
    return Incomplete;

  message->body_.append(chunkData.rawText(), size - message->body_.size());

  return Complete;
}

State BodyStep::parseTransferEncoding(
    StreamCursor &cursor, const std::shared_ptr<Header::TransferEncoding> &te) {
  auto encoding = te->encoding();
  if (encoding == Http::Header::Encoding::Chunked) {
    Chunk::Result result;
    try {
      while ((result = chunk.parse(cursor)) != Chunk::Final) {
        if (result == Chunk::Incomplete)
          return State::Again;

        chunk.reset();
        if (cursor.eof())
          return State::Again;
      }
      chunk.reset();
    } catch (const std::exception &e) {
      // reset chunk incase signal handled & chunk eventually reused
      chunk.reset();
      raise(e.what());
    }

    return State::Done;
  } else {
    raise("Unsupported Transfer-Encoding", Code::Not_Implemented);
  }
  return State::Done;
}

ParserBase::ParserBase(size_t maxDataSize)
    : buffer(maxDataSize), cursor(&buffer) {}

State ParserBase::parse() {
  State state;
  do {
    Step *step = allSteps[currentStep].get();
    state = step->apply(cursor);
    if (state == State::Next) {
      ++currentStep;
    }
  } while (state == State::Next);

  // Should be either Again or Done
  return state;
}

bool ParserBase::feed(const char *data, size_t len) {
  return buffer.feed(data, len);
}

void ParserBase::reset() {
  buffer.reset();
  cursor.reset();

  currentStep = 0;
}

} // namespace Private

namespace Uri {

Query::Query() : params() {}

Query::Query(
    std::initializer_list<std::pair<const std::string, std::string>> params)
    : params(params) {}

void Query::add(std::string name, std::string value) {
  params.insert(std::make_pair(std::move(name), std::move(value)));
}

Optional<std::string> Query::get(const std::string &name) const {
  auto it = params.find(name);
  if (it == std::end(params))
    return Optional<std::string>(None());

  return Optional<std::string>(Some(it->second));
}

std::string Query::as_str() const {
  std::string query_url;
  for (const auto &e : params) {
    query_url += "&" + e.first + "=" + e.second;
  }
  if (!query_url.empty()) {
    query_url[0] = '?'; // replace first `&` with `?`
  }
  return query_url;
}

bool Query::has(const std::string &name) const {
  return params.find(name) != std::end(params);
}

} // namespace Uri

Message::Message(Version version) : version_(version) {}

Version Message::version() const { return version_; }

Code Message::code() const { return code_; }

std::string Message::body() const { return body_; }

const Header::Collection &Message::headers() const { return headers_; }

Header::Collection &Message::headers() { return headers_; }

const CookieJar &Message::cookies() const { return cookies_; }

CookieJar &Message::cookies() { return cookies_; }

Method Request::method() const { return method_; }

const std::string &Request::resource() const { return resource_; }

const Uri::Query &Request::query() const { return query_; }

const Address &Request::address() const { return address_; }

std::chrono::milliseconds Request::timeout() const { return timeout_; }

Response::Response(Version version) : Message(version) {}

#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
std::shared_ptr<Tcp::Peer> Request::peer() const {
  auto p = peer_.lock();

  if (!p)
    throw std::runtime_error("Failed to retrieve peer: Broken pipe");

  return p;
}
#endif

ResponseStream::ResponseStream(ResponseStream &&other)
    : response_(std::move(other.response_)), peer_(std::move(other.peer_)),
      buf_(std::move(other.buf_)), transport_(other.transport_),
      timeout_(std::move(other.timeout_)) {}

ResponseStream::ResponseStream(Message &&other, std::weak_ptr<Tcp::Peer> peer,
                               Tcp::Transport *transport, Timeout timeout,
                               size_t streamSize, size_t maxResponseSize)
    : response_(std::move(other)), peer_(std::move(peer)),
      buf_(streamSize, maxResponseSize), transport_(transport),
      timeout_(std::move(timeout)) {
  if (!writeStatusLine(response_.version(), response_.code(), buf_))
    throw Error("Response exceeded buffer size");

  if (!writeCookies(response_.cookies(), buf_)) {
    throw Error("Response exceeded buffer size");
  }

  if (writeHeaders(response_.headers(), buf_)) {
    std::ostream os(&buf_);
    /* @Todo @Major:
     * Correctly handle non-keep alive requests
     * Do not put Keep-Alive if version == Http::11 and request.keepAlive ==
     * true
     */
    // writeHeader<Header::Connection>(os, ConnectionControl::KeepAlive);
    // if (!os) throw Error("Response exceeded buffer size");
    writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked);
    if (!os)
      throw Error("Response exceeded buffer size");
    os << crlf;
  }
}

ResponseStream &ResponseStream::operator=(ResponseStream &&other) {
  response_ = std::move(other.response_);
  peer_ = std::move(other.peer_);
  buf_ = std::move(other.buf_);
  transport_ = other.transport_;
  timeout_ = std::move(other.timeout_);

  return *this;
}

std::streamsize ResponseStream::write(const char *data, std::streamsize sz) {
  std::ostream os(&buf_);
  os << std::hex << sz << crlf;
  os.write(data, sz);
  os << crlf;
  return sz;
}

std::shared_ptr<Tcp::Peer> ResponseStream::peer() const {
  if (peer_.expired()) {
    throw std::runtime_error("Write failed: Broken pipe");
  }

  return peer_.lock();
}

void ResponseStream::flush() {
  timeout_.disarm();
  auto buf = buf_.buffer();

  auto fd = peer()->fd();
  transport_->asyncWrite(fd, buf);
  transport_->flush();

  buf_.clear();
}

void ResponseStream::ends() {
  std::ostream os(&buf_);
  os << "0" << crlf;
  os << crlf;

  if (!os) {
    throw Error("Response exceeded buffer size");
  }

  flush();
}

ResponseWriter::ResponseWriter(ResponseWriter &&other)
    : response_(std::move(other.response_)), peer_(other.peer_),
      buf_(std::move(other.buf_)), transport_(other.transport_),
      timeout_(std::move(other.timeout_)), sent_bytes_(0) {}

ResponseWriter::ResponseWriter(Tcp::Transport *transport, Request request,
                               Handler *handler, std::weak_ptr<Tcp::Peer> peer)
    : response_(request.version()), peer_(peer),
      buf_(DefaultStreamSize, handler->getMaxResponseSize()),
      transport_(transport),
      timeout_(transport, handler, std::move(request), peer), sent_bytes_(0) {}

ResponseWriter::ResponseWriter(const ResponseWriter &other)
    : response_(other.response_), peer_(other.peer_),
      buf_(DefaultStreamSize, other.buf_.maxSize()),
      transport_(other.transport_), timeout_(other.timeout_), sent_bytes_(0) {}

void ResponseWriter::setMime(const Mime::MediaType &mime) {
  auto ct = response_.headers().tryGet<Header::ContentType>();
  if (ct) {
    ct->setMime(mime);
  } else {
    response_.headers().add(std::make_shared<Header::ContentType>(mime));
  }
}

Async::Promise<ssize_t> ResponseWriter::sendMethodNotAllowed(
    const std::vector<Http::Method> &supportedMethods) {
  response_.code_ = Http::Code::Method_Not_Allowed;
  response_.headers().add(
      std::make_shared<Http::Header::Allow>(supportedMethods));
  const std::string &body =
      codeString(Pistache::Http::Code::Method_Not_Allowed);
  return putOnWire(body.c_str(), body.size());
}

Async::Promise<ssize_t> ResponseWriter::send(Code code, const std::string &body,
                                             const Mime::MediaType &mime) {
  return sendImpl(code, body.c_str(), body.size(), mime);
}

Async::Promise<ssize_t> ResponseWriter::send(Code code, const char *data,
                                             const size_t size,
                                             const Mime::MediaType &mime) {
  return sendImpl(code, data, size, mime);
}

Async::Promise<ssize_t> ResponseWriter::sendImpl(Code code, const char *data,
                                                 const size_t size,
                                                 const Mime::MediaType &mime) {
  response_.code_ = code;

  if (mime.isValid()) {
    auto contentType = headers().tryGet<Header::ContentType>();
    if (contentType) {
      contentType->setMime(mime);
    } else {
      headers().add(std::make_shared<Header::ContentType>(mime));
    }
  }

  return putOnWire(data, size);
}

ResponseStream ResponseWriter::stream(Code code, size_t streamSize) {
  response_.code_ = code;

  return ResponseStream(std::move(response_), peer_, transport_,
                        std::move(timeout_), streamSize, buf_.maxSize());
}

const CookieJar &ResponseWriter::cookies() const { return response_.cookies(); }

CookieJar &ResponseWriter::cookies() { return response_.cookies(); }

const Header::Collection &ResponseWriter::headers() const {
  return response_.headers();
}

Header::Collection &ResponseWriter::headers() { return response_.headers(); }

Timeout &ResponseWriter::timeout() { return timeout_; }

std::shared_ptr<Tcp::Peer> ResponseWriter::peer() const {
  if (peer_.expired()) {
    throw std::runtime_error("Write failed: Broken pipe");
  }

  return peer_.lock();
}

DynamicStreamBuf *ResponseWriter::rdbuf() { return &buf_; }

DynamicStreamBuf *ResponseWriter::rdbuf(DynamicStreamBuf *other) {
  UNUSED(other)
  throw std::domain_error("Unimplemented");
}

ResponseWriter ResponseWriter::clone() const { return ResponseWriter(*this); }

Async::Promise<ssize_t> ResponseWriter::putOnWire(const char *data,
                                                  size_t len) {
  try {
    std::ostream os(&buf_);

#define OUT(...)                                                               \
  do {                                                                         \
    __VA_ARGS__;                                                               \
    if (!os) {                                                                 \
      return Async::Promise<ssize_t>::rejected(                                \
          Error("Response exceeded buffer size"));                             \
    }                                                                          \
  } while (0);

    OUT(writeStatusLine(response_.version(), response_.code(), buf_));
    OUT(writeHeaders(response_.headers(), buf_));
    OUT(writeCookies(response_.cookies(), buf_));

    /* @Todo @Major:
     * Correctly handle non-keep alive requests
     * Do not put Keep-Alive if version == Http::11 and request.keepAlive ==
     * true
     */
    // OUT(writeHeader<Header::Connection>(os, ConnectionControl::KeepAlive));
    OUT(writeHeader<Header::ContentLength>(os, len));

    OUT(os << crlf);

    if (len > 0) {
      OUT(os.write(data, len));
    }

    auto buffer = buf_.buffer();
    sent_bytes_ += buffer.size();

    timeout_.disarm();

#undef OUT

    auto fd = peer()->fd();

    return transport_->asyncWrite(fd, buffer)
        .then<std::function<Async::Promise<ssize_t>(ssize_t)>,
              std::function<void(std::exception_ptr &)>>(
            [=](int /*l*/) {
              return Async::Promise<ssize_t>([=](
                  Async::Deferred<ssize_t> /*deferred*/) mutable { return; });
            },

            [=](std::exception_ptr &eptr) {
              return Async::Promise<ssize_t>::rejected(eptr);
            });

  } catch (const std::runtime_error &e) {
    return Async::Promise<ssize_t>::rejected(e);
  }
}

Async::Promise<ssize_t> serveFile(ResponseWriter &writer,
                                  const std::string &fileName,
                                  const Mime::MediaType &contentType) {
  struct stat sb;

  int fd = open(fileName.c_str(), O_RDONLY);
  if (fd == -1) {
    std::string str_error(strerror(errno));
    if (errno == ENOENT) {
      throw HttpError(Http::Code::Not_Found, std::move(str_error));
    }
    // eles if TODO
    /* @Improvement: maybe could we check for errno here and emit a different
       error message
    */
    else {
      throw HttpError(Http::Code::Internal_Server_Error, std::move(str_error));
    }
  }

  int res = ::fstat(fd, &sb);
  close(fd); // Done with fd, close before error can be thrown
  if (res == -1) {
    throw HttpError(Code::Internal_Server_Error, "");
  }

  auto *buf = writer.rdbuf();

  std::ostream os(buf);

#define OUT(...)                                                               \
  do {                                                                         \
    __VA_ARGS__;                                                               \
    if (!os) {                                                                 \
      return Async::Promise<ssize_t>::rejected(                                \
          Error("Response exceeded buffer size"));                             \
    }                                                                          \
  } while (0);

  auto setContentType = [&](const Mime::MediaType &contentType) {
    auto &headers = writer.headers();
    auto ct = headers.tryGet<Header::ContentType>();
    if (ct)
      ct->setMime(contentType);
    else
      headers.add<Header::ContentType>(contentType);
  };

  OUT(writeStatusLine(writer.response_.version(), Http::Code::Ok, *buf));
  if (contentType.isValid()) {
    setContentType(contentType);
  } else {
    auto mime = Mime::MediaType::fromFile(fileName.c_str());
    if (mime.isValid())
      setContentType(mime);
  }

  OUT(writeHeaders(writer.headers(), *buf));

  const size_t len = sb.st_size;

  OUT(writeHeader<Header::ContentLength>(os, len));

  OUT(os << crlf);

  auto *transport = writer.transport_;
  auto peer = writer.peer();
  auto sockFd = peer->fd();

  auto buffer = buf->buffer();
  return transport->asyncWrite(sockFd, buffer, MSG_MORE)
      .then(
          [=](ssize_t) {
            return transport->asyncWrite(sockFd, FileBuffer(fileName));
          },
          Async::Throw);

#undef OUT
}

Private::ParserImpl<Http::Request>::ParserImpl(size_t maxDataSize)
    : ParserBase(maxDataSize), request() {
  allSteps[0].reset(new RequestLineStep(&request));
  allSteps[1].reset(new HeadersStep(&request));
  allSteps[2].reset(new BodyStep(&request));
}

void Private::ParserImpl<Http::Request>::reset() {
  ParserBase::reset();

  request = Request();
}

Private::ParserImpl<Http::Response>::ParserImpl(size_t maxDataSize)
    : ParserBase(maxDataSize), response() {
  allSteps[0].reset(new ResponseLineStep(&response));
  allSteps[1].reset(new HeadersStep(&response));
  allSteps[2].reset(new BodyStep(&response));
}

void Handler::onInput(const char *buffer, size_t len,
                      const std::shared_ptr<Tcp::Peer> &peer) {
  auto &parser = getParser(peer);
  try {
    if (!parser.feed(buffer, len)) {
      parser.reset();
      throw HttpError(Code::Request_Entity_Too_Large,
                      "Request exceeded maximum buffer size");
    }

    auto state = parser.parse();

    if (state == Private::State::Done) {
      ResponseWriter response(transport(), parser.request, this, peer);

#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
      parser.request.associatePeer(peer);
#endif

      auto request = parser.request;
      request.copyAddress(peer->address());

      auto connection = request.headers().tryGet<Header::Connection>();

      if (connection) {
        response.headers().add<Header::Connection>(connection->control());
      } else {
        response.headers().add<Header::Connection>(ConnectionControl::Close);
      }

      onRequest(request, std::move(response));
      parser.reset();
    }

  } catch (const HttpError &err) {
    ResponseWriter response(transport(), parser.request, this, peer);
    response.send(static_cast<Code>(err.code()), err.reason());
    parser.reset();
  }

  catch (const std::exception &e) {
    ResponseWriter response(transport(), parser.request, this, peer);
    response.send(Code::Internal_Server_Error, e.what());
    parser.reset();
  }
}

void Handler::onConnection(const std::shared_ptr<Tcp::Peer> &peer) {
  peer->putData(ParserData, std::make_shared<RequestParser>(maxRequestSize_));
}

void Handler::onDisconnection(const std::shared_ptr<Tcp::Peer> & /*peer*/) {}

void Handler::onTimeout(const Request & /*request*/,
                        ResponseWriter /*response*/) {}

Timeout::~Timeout() { disarm(); }

void Timeout::disarm() {
  if (transport && armed) {
    transport->disarmTimer(timerFd);
  }
}

bool Timeout::isArmed() const { return armed; }

Timeout::Timeout(Tcp::Transport *transport_, Handler *handler_,
                 Request request_, std::weak_ptr<Tcp::Peer> peer_)
    : handler(handler_), request(std::move(request_)), transport(transport_),
      armed(false), timerFd(-1), peer(peer_) {}

void Timeout::onTimeout(uint64_t numWakeup) {
  UNUSED(numWakeup)
  if (!peer.lock())
    return;

  ResponseWriter response(transport, request, handler, peer);

  handler->onTimeout(request, std::move(response));
}

void Handler::setMaxRequestSize(size_t value) { maxRequestSize_ = value; }

size_t Handler::getMaxRequestSize() const { return maxRequestSize_; }

void Handler::setMaxResponseSize(size_t value) { maxResponseSize_ = value; }

size_t Handler::getMaxResponseSize() const { return maxResponseSize_; }

RequestParser &
Handler::getParser(const std::shared_ptr<Tcp::Peer> &peer) const {
  return static_cast<RequestParser &>(*peer->getData(ParserData));
}

} // namespace Http
} // namespace Pistache
